package cn.gwm.flink.streaming.function.process;

import cn.gwm.flink.streaming.client.RedisClient;
import cn.gwm.flink.streaming.constant.ChargePredictFields;
import cn.gwm.flink.streaming.constant.DcBatterMapVal;
import cn.gwm.flink.streaming.strategy.vehiclepredict.*;
import cn.gwm.flink.streaming.task.ChargePredictDwsTask;
import cn.gwm.utils.ConfigLoader;
import cn.gwm.utils.HBaseDataSource;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.MapTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import redis.clients.jedis.JedisCluster;

import java.sql.Connection;
import java.util.Map;

/**
 * @author : GW00280745
 * @date : 2023/1/15 15:34
 * @description :
 */
@Slf4j
public class ChargePredictProcessFunction extends KeyedBroadcastProcessFunction<Object, JSONObject, JSONObject, JSONObject> {
    @Override
    public void processElement(JSONObject line, KeyedBroadcastProcessFunction<Object, JSONObject, JSONObject, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //1.求充电组名和初始点
        chargeId.process(line,chargeBeginState,latestTidState,startPointState,latestBatterState);
        //对one soc 的time进行记录
        oneSocState.process(line,oneSocTimeState);
        if (ChargePredictDwsTask.predict_type==1){
            //2.求首条记录点
            firstRecord.process(line,frState);
            //3.求温升map表
            mapTempInterval.process(line,mapTempIntervalState,ctx);
            mapTempAvg.process(line,mapTempAvgState,ctx,mapTempAvgJedis);
            //4.求电流map表
            mapBatterInterval.process(line,mapBatterIntervalState,ctx);
            mapBatterAvg.process(line,mapBatterAvgState,ctx,mapBatterAvgJedis,mapTempAvgState);
        }
        //5.求预测值
        if (!ChargePredictDwsTask.predict_type_1_only_map || ChargePredictDwsTask.predict_type==num2){
            socInterval.process(line, latestSocIntervalPointState,latestEstimateState,latestBatterState,ctx,mapTempAvgState,socIntervalJedis,oneSocTimeState);
        }
        //other
        //end.collect
        out.collect(line);
    }
    private int num2=2;
    @Override
    public void processBroadcastElement(JSONObject confParam, KeyedBroadcastProcessFunction<Object, JSONObject, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        ctx.getBroadcastState(confInfoMsd).put(null,confParam);
    }
    private ChargeId chargeId;
    private FirstRecord firstRecord;
    private MapTempInterval mapTempInterval;
    private MapTempAvg mapTempAvg;
    private MapBatterInterval mapBatterInterval;
    private MapBatterAvg mapBatterAvg;
    private SocInterval socInterval;
    private OneSocState oneSocState;
    private String onlyMap;
    private JedisCluster mapTempAvgJedis;
    private JedisCluster mapBatterAvgJedis;
    private JedisCluster socIntervalJedis;
    private MapStateDescriptor<String, Map<String, Object>> confInfoMsd = new MapStateDescriptor<>("conf-info", BasicTypeInfo.STRING_TYPE_INFO, new MapTypeInfo<>(String.class, Object.class));

    public ChargePredictProcessFunction(){
        // onlyMap = ChargePredictFields.ins().confInfo.getOrDefault("only.process.map_table","false")
        onlyMap = String.valueOf(ChargePredictDwsTask.predict_type_1_only_map);
        //充电初始化
        String diffStr = ChargePredictFields.ins().confInfo.getOrDefault("two.charge.diff", "30");
        Integer chargeSeparatorTime = Integer.parseInt(diffStr) * 60 * 1000;
        Integer iniPointBatterVal = -10;
        this.chargeId = new ChargeId(iniPointBatterVal, chargeSeparatorTime);
        this.firstRecord = new FirstRecord();
        this.mapTempInterval=new MapTempInterval();
        //MapTemp
        this.mapTempAvg = new MapTempAvg("dcp:t3:");
        //MapBatter
        this.mapBatterInterval = new MapBatterInterval("10");
        this.mapBatterAvg = new MapBatterAvg();
        //SocInterval
        JSONObject defaultBatterMap = dcBatterMapDefault();
        SocMapDefault socMapDefault = new SocMapDefault(defaultBatterMap);
        this.socInterval = new SocInterval("dcp:t3:","dcp:b3:",defaultBatterMap,socMapDefault);
        this.oneSocState = new OneSocState();
    }
    private JSONObject dcBatterMapDefault(){
        DcBatterMapVal mapVal = new DcBatterMapVal();
        JSONObject jsonObject = mapVal.load();
        return jsonObject;
    }
    private MapState<String, String> chargeBeginState;
    private MapState<String, Long> latestTidState;
    private MapState<String, String> startPointState;
    private MapState<String, String> frState;
    private MapState<String, String> mapTempIntervalState;
    private MapState<String, Map> mapTempAvgState;
    private MapState<String, String> mapBatterIntervalState;
    private MapState<String, String> mapBatterAvgState;
    private MapState<String, String> latestSocIntervalPointState;
    private MapState<String, String> latestEstimateState;
    private MapState<String, String> latestBatterState;
    private MapState<String, String> oneSocTimeState;
    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化一堆状态
        chargeBeginState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("chargeBeginState", String.class, String.class));
        latestTidState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Long>("latestTidState", String.class, Long.class));
        startPointState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("startPointState", String.class, String.class));
        frState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("frState", String.class, String.class));
        mapTempIntervalState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("mapTempIntervalState", String.class, String.class));
        mapTempAvgState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Map>("mapTempAvgState", String.class, Map.class));
        mapBatterIntervalState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("mapBatterIntervalState", String.class, String.class));
        mapBatterAvgState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("mapBatterAvgState", String.class, String.class));
        latestSocIntervalPointState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("latestSOCIntervalPointState", String.class, String.class));
        latestEstimateState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("latestEstimateState", String.class, String.class));
        latestBatterState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("latestBatterState", String.class, String.class));
        oneSocTimeState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("oneSocTimeState", String.class, String.class));
        JedisCluster jedisCluster = null;
//        JedisCluster jedisCluster = RedisInfo.getInstance().jedisPoolAli()
//        JedisCluster jedisCluster = RedisInfo.getInstance().jedisPoolInnerNet()
        try {
            //南方提供的redis客户端

            String redisCluster = ConfigLoader.get("redis.cluster");
            String password = ConfigLoader.get("redis.cluster.password");
            System.out.println("redis_cluster:"+redisCluster+",pwd:"+password);
            jedisCluster = RedisClient.jedisCluster;
            mapTempAvgJedis = jedisCluster;
            mapBatterAvgJedis = jedisCluster;
            socIntervalJedis = jedisCluster;
        }catch (Exception e){
            e.printStackTrace();
        }
        //redis的数据遍历一次将其写入rocks
        if (ChargePredictDwsTask.predict_type==1){
            //这一步比较耗时
            RocksData.obj().redis2Rocks(jedisCluster,mapTempAvgState);
        }else {
            try {
                Connection connection = HBaseDataSource.ins().getDataSource().getConnection();
                HBaseDataSource.ins().close(connection);
            }catch (Exception e){
                log.error(ExceptionUtils.getFullStackTrace(e));
            }
        }
    }


}
